/**
 * 
 */
package jvs.peer.io;

import java.io.IOException;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;

import jvs.peer.misc.Base64;
import jvs.peer.util.Log;


/**
 * @author qiangli
 *
 */
public class MessageOutputStream extends OutputStream {
	private Channel channel = null;

	private byte[] buf = new byte[MessageConstant.MAX_MESSAGE_SIZE];
	private int pos = 0;
	private int seqno = 0; //messages sent
	private boolean closed = false;
	
	private int counter = 0; //bytes sent
	
	public MessageOutputStream(Channel channel) {
		this.channel = channel;
	}

	@Override
	public void write(int b) throws IOException {
		buf[pos++] = (byte)b;
		if (pos >= buf.length) {
			send(buf,buf.length,MessageConstant.data);
			pos = 0;
		}
	}

	@Override
	public void close() throws IOException {
		if (closed) {
			return;
		}
		closed = true;
		flush();
		send(MessageConstant.close);
		//
		this.buf = null;
		this.channel = null;
	}

	@Override
	public void flush() throws IOException {
		if (pos > 0) {
			send(buf,pos,MessageConstant.data);
			pos = 0;
		}
	}
	
	public void open() throws IOException {
		send(MessageConstant.open);
	}
	
	protected void send(int ioscmd) throws IOException {
		send(null,0,ioscmd);
	}
	
	protected void send(byte[] buf, int len, int ioscmd) throws IOException {
		try {
			Message msg = new Message();
			msg.setIoscmd(ioscmd);
			msg.setData(encode(buf,len,"UTF-8"));
			msg.setSeqno(seqno);
	
			channel.send(msg);	
			seqno++;
			
			counter += len;
			
			if(Log.DEBUG) {
				Log.log(this, Thread.currentThread().getId()+"@@@@ "+channel.getId()+" out seqno: "+seqno+" counter: "+counter);
			}
		} catch (Exception e) {
			e.printStackTrace();
			throw new IOException(e.getMessage());
		}
	}


	private String encode(byte[] buf, int len, String charset) throws UnsupportedEncodingException {
		if (len <=0 ) {
			return null;
		}
		return Base64.encodeBytes(buf,0,len);
	}
}
